package com.tensai.study.framework.reactornetty;

import io.netty.util.concurrent.DefaultThreadFactory;
import lombok.extern.slf4j.Slf4j;
import org.junit.Test;
import reactor.core.Exceptions;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.netty.tcp.TcpClient;
import reactor.netty.tcp.TcpSslContextSpec;
import reactor.util.retry.Retry;

import java.time.Duration;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.stream.IntStream;

/**
 * @author Tensai
 */
@Slf4j
public class MyTcpClientTest {

    @Test
    public void test() throws InterruptedException {

        CountDownLatch latch = new CountDownLatch(1);
        StringBuilder toSend = new StringBuilder("a");

        TcpClient.create() // 准备要配置的TCP客户端
                 // 服务端口
                 .port(11551)
                 // 配置SSL，以提供已配置的SslContext。
//                 .secure(spec -> spec
//                         .sslContext(TcpSslContextSpec.forClient()))
                 //.wiretap()
//                 .doOnConnected(con -> {
//                     log.info("客户端链接成功");
//                     // 下一个导致Publisher <Void>的运算符的下一个序列永远不会自己完成，因此客户端保持永久连接
//                     con.outbound()
//                        .sendString(Mono.just(toSend.toString()))
//                        .then(con.inbound()
//                                 .receive()
//                                 .asString()
//                                 .log("TCP-连接")
//                                 .doOnNext(s -> log.info("服务器返回: " + s))
//                                 .flatMap(s -> con.outbound()
//                                                  .sendString(Mono.just(toSend.append("a").toString()))
//                                                  .then()
//                                 )
//                        )
//                        .then()
//                        .subscribe();
//                 })
                 .doOnDisconnected(con -> {
                     log.info("Server disconnected!");
                     latch.countDown();
                 })
                 .connect()
                 // 方式1：简单重试，retry()动作是当操作序列发生错误后重新订阅序列。
                 // 重试3次
                 .retry(3)

                 // 方式2：backoff方法返回就其实是Retry的子类RetryBackoffSpec。它需要两个参数：最大重试次数和最小间隔时间。
                 // 最多重试3次，每次的最短时间间隔为5秒
                 .retryWhen(Retry.backoff(3, Duration.ofSeconds(5)))

                 // 方式3：fixedDelay方法返回的也是RetryBackoffSpec。它需要两个参数:最大重试次数和固定的间隔时间。
                 // 最大重试3次，固定延迟5秒
                 .retryWhen(Retry.fixedDelay(3, Duration.ofSeconds(5)))

                 // 方式4：from方法。它需要一个Function函数。
                 .retryWhen(Retry.from((retrySignals) -> retrySignals.map(this::getNumberOfTries)))

                 // 方式5：withThrowable方法。它和from有一点类似也是接收一个Function,但是它的参数是异常。
                 .retryWhen(Retry.withThrowable((retrySignals) -> retrySignals.map(rs -> {
                     if (rs instanceof Exception) {
                         throw new RuntimeException("重试错误");
                     } else {
                         return rs;
                     }
                 })))
                 .log("TCP-客户端")
                 .doOnError(e -> log.error("服务器链接失败 ... " + e.getMessage()))
                 //.retryBackoff(Long.MAX_VALUE, Duration.ofSeconds(3), Duration.ofSeconds(10)) // 重试服务
                .handle((t, s) -> {

                })
                 .block();

        latch.await(); // 客户端正在运行，直到服务器断开与客户端的连接
    }


    private Long getNumberOfTries(Retry.RetrySignal rs) {
        log.info("重试：" + rs.totalRetries());
        if (rs.totalRetries() < 3) {
            return rs.totalRetries();
        } else {
            log.error("retries exhausted");
            throw Exceptions.propagate(rs.failure());
        }
    }

    @Test
    public void test1() throws InterruptedException {
        ThreadPoolExecutor tpe = new ThreadPoolExecutor(Integer.MAX_VALUE, Integer.MAX_VALUE,
                                                        100, TimeUnit.MILLISECONDS, new ArrayBlockingQueue<>(5),
                                                        new DefaultThreadFactory("test-"));
        Flux.interval(Duration.ofMillis(100))
                 .subscribe(it -> {
                     tpe.execute(() -> {
                         try {
                             test();
                             Thread.sleep(20);
                         } catch (InterruptedException e) {
                             throw new RuntimeException(e);
                         }
                     });
                 });

        Thread.sleep(600000);
    }

    @Test
    public void test2() throws InterruptedException {
        test1();
    }
}
